
#include <iostream>
#include <string>
#include <sstream>
#include <TRandom.h>
#include "json/reader.h"
#include "json/writer.h"
#include "json/value.h"
#include "GBGwServer.h"
#include "GBGwStreamSource.h"
#include "SDRedisHelper.h"
#define LOG_TAG "GBGwServer"
//#define WITH_MQ_AUTH
#define MQSERVERPASS "9845544bd8900df1707268ea6bc056d8"

static gbt::GbtProxyAgent& getAgent()
{
    return gbt::GbtProxyAgent::instance();
}



GBGwServer::GBGwServer()
: _subAborth(false), _connAbort(false)
, _connRun(false), _subRun(false),_workRun(false),_pubRun(nullptr)
,_workThread(nullptr),_subThread(nullptr),_pubThread(nullptr),_connThread(nullptr)
{
    comn::Random::makeSeed();
    _terminal = std::make_shared<RelayTerminal>(this,this);
    _index = 0;
}

GBGwServer::~GBGwServer()
{
    stopServer();
    _terminal.reset();
}


void GBGwServer::startServer(struct GBRelayConfig &config)
{
    _redis_port = config.redis_port;
    _redis_ip = std::string(config.redis_ip);

    _media_port = config.inner_media_port;
    _media_ip = std::string(config.inner_media_ip);

    //start sip agent
    {
        m_server.port = config.upper_server_port;
        strcpy(m_server.ip,config.upper_server_ip);

        strcpy(m_server.id,config.sip_username); //Server Domain
        strcpy(m_server.username,config.sip_username); //User
        strcpy(m_server.password,config.sip_password); //Password
        getAgent().setTransport(GBT_TRANSPORT_UDP);

        getAgent().setSink(this);
        getAgent().setMessageHandler(this);
        getAgent().setPlaybackHandler(this);

        getAgent().setRegTime(config.upper_reg_interval);
        getAgent().setUserAgent("EasyGbSvr");
        getAgent().setHeartBeat(config.upper_refresh_interval, 3);

        GBT_Catalog catalog = GBT_Catalog();
        strcpy(catalog.name, "EasyGbSvr");
        getAgent().setCatalog(catalog);
        getAgent().open("0.0.0.0",config.inner_sip_port, m_server);
    }

    {
        Json::Reader reader;
        Json::Value exports;

        sw::redis::ConnectionOptions connection_options;
        connection_options.host = _redis_ip;  // Required.
        connection_options.port = _redis_port; // Optional. The default port is 6379.
        connection_options.password = MQSERVERPASS;
        connection_options.socket_timeout = std::chrono::milliseconds(REDIS_TIMEDOUT_MS);
        auto redis_client = std::make_shared<sw::redis::Redis>(connection_options);

        auto gb_exported = redisXXGetStringLock(redis_client,"prop.secure.gb28181.lock","prop.persist.device.gb28181");

         if (reader.parse(gb_exported, exports) && exports.isArray()) {
            std::unique_lock<std::mutex> vlck(_vlock);
            std::unique_lock<std::mutex> alck(_alock);
            for (auto device : exports)
            {
                auto obj = std::make_shared<ProxyObj>();
                obj->uid = device["gid"].asInt();
                //if((obj->uid != 1000003061) && (obj->uid != 1000001140) && (obj->uid != 1000002129) && (1000000426 != obj->uid))
                //    continue;
                obj->on_pos = false;
                obj->living = false;
                obj->channel_online = false;
                obj->index = _index++;
                obj->name = device["dev_alias"].asString();
                obj->gb28181_id = device["gb28181_id"].asUInt();
                _proxyObjs[obj->uid] = obj;
            }
        }
    }

#if 0
    {
        auto obj = std::make_shared<ProxyObj>();
        obj->uid = 1000000005;
        obj->index = _proxyObjs.size();
        obj->name = std::string("设备10001");
        _proxyObjs[obj->uid] = obj;;
    }
#endif
    //create channel for all exports
    {
        std::unique_lock<std::mutex> vlck(_vlock);
        std::unique_lock<std::mutex> alck(_alock);
        for(auto iter = _proxyObjs.begin(); iter != _proxyObjs.end();iter++)
        {
            auto second = iter->second;
            char deviceID[GBT_UNICODE_LENGTH + 1] = { 0 };
            //340200000013255 00002
            memcpy(deviceID, m_server.username,10); 
            snprintf(deviceID + 10,sizeof(deviceID) - 10,"13255%05d", second->gb28181_id);

    
            //memcpy(deviceID, m_server.username,10); 
            //snprintf(deviceID + 10,sizeof(deviceID) - 10,"132%07d", second->gb28181_id);

            second->channel = std::make_shared<GBGwChannel>(this,second->name,m_server.username,deviceID,second->index);
            std::cout << "new obj[" << second->index << "] uid:"<< second->uid << ",name:" << second->name  << ",gb28181_id:" << second->gb28181_id << ",deviceID:"<< deviceID << ",channel:" << second->channel<< std::endl;       
        }
    }

    {
        //start redis subscription
        std::unique_lock<std::mutex> lck(_subLock);
        if(_subRun)
            return;
        _subAborth = false;
        _subThread = std::make_shared<std::thread>([&](){
            subThreadEntry();
        });
        _subCond.wait(lck);
    }
#if 1

    {
        //start redis subscription
        std::unique_lock<std::mutex> lck(_workLock);
        if(_workRun)
            return;
        _workAborth = false;
        _workThread = std::make_shared<std::thread>([&](){
            workThreadEntry();
        });
        _workCond.wait(lck);
    }



    //start connect mserver
    {
        std::unique_lock<std::mutex> lck(_connLock);
        if(_connRun)
            return;
        _connAbort = false;
        _connThread = std::make_shared<std::thread>([&](){
            connThreadEntry();
        });
        _connCond.wait(lck);
    }


    {
        //start redis subscription
        std::unique_lock<std::mutex> lck(_pubLock);
        if(_pubRun)
            return;
        _pubAborth = false;
        _pubThread = std::make_shared<std::thread>([&](){
            pubThreadEntry();
        });
        _pubCond.wait(lck);
    }
#endif

}

void GBGwServer::stopServer()
{
    //停止REDIS消息订阅
    {
        std::unique_lock<std::mutex> lck(_subLock);
        if(_subRun && _subThread){
            _subAborth = true;
            _subCond.notify_all();
            _subThread->join();
            _subThread = nullptr;
        }
    }

    {
        std::unique_lock<std::mutex> lck(_workLock);
        if(_workRun && _workThread){
            _workAborth = true;
            _workCond.notify_all();
            _workThread->join();
            _workThread = nullptr;
        }
    }

    //stop mserver connection
    {
        std::unique_lock<std::mutex> lck(_connLock);
        if (_subRun && _connThread) {
            _connAbort = true;
            _connCond.notify_all();
            _connThread->join();
            _connThread = nullptr;
        }
    }

    //close sip agent
    getAgent().close();

    {
        //先停止接收音视频，防止音视频回调竞争锁
        std::unique_lock<std::mutex> vlck(_vlock);
        std::unique_lock<std::mutex> alck(_alock);
        for(auto iter = _proxyObjs.begin();iter != _proxyObjs.end();iter++)
        {
            auto second = iter->second;
            _terminal->offPosition(second->uid);
            second->channel->offline();
        }
        _proxyObjs.clear();
    }
}

void GBGwServer::postPubMessage(std::string key,std::string value)
{
    std::unique_lock<std::mutex> lk(_pubLock);
    if(!_pubRun)
        return;
    _lstpubMsg.push_back(std::make_pair(key,value));
    _pubCond.notify_all();
}


void GBGwServer::pubThreadEntry()
{
    redisContext  *ctx = nullptr;
    std::list<std::pair<std::string,std::string>> localPubMsg;


    sw::redis::ConnectionOptions connection_options;
    connection_options.host = _redis_ip;  // Required.
    connection_options.port = _redis_port; // Optional. The default port is 6379.
    connection_options.password = MQSERVERPASS;
    connection_options.socket_timeout = std::chrono::milliseconds(REDIS_TIMEDOUT_MS);
    auto redis_client = std::make_shared<sw::redis::Redis>(connection_options);


    {
        std::unique_lock<std::mutex> lk(_pubLock);
         _pubRun = true;
        _pubCond.notify_all();
    }

    do 
    {

        if(localPubMsg.size() <= 0){
            std::unique_lock<std::mutex> lk(_pubLock);
            if(_lstpubMsg.size() <= 0){
                _pubCond.wait(lk);
                continue;
            }
            localPubMsg = std::move(_lstpubMsg);
            continue;
        }

        for(auto msg : localPubMsg)
            redis_client->publish(msg.first,msg.second);
        localPubMsg.clear();
    }while(!_workAborth);
    if(ctx)
        redisFree(ctx);

    {
        std::unique_lock<std::mutex> lk(_pubLock);
        _pubCond.notify_all();
        _pubRun = false;
    }
}

void GBGwServer::workThreadEntry()
{
    redisContext  *ctx = nullptr;
    std::list<std::pair<std::string,std::string>> localWorkMsg;
    time_t t_last_get = 0 ,t_now = 0;
    

    sw::redis::ConnectionOptions connection_options;
    connection_options.host = _redis_ip;  // Required.
    connection_options.port = _redis_port; // Optional. The default port is 6379.
    connection_options.password = MQSERVERPASS;
    connection_options.socket_timeout = std::chrono::milliseconds(REDIS_TIMEDOUT_MS);
    auto redis_client = std::make_shared<sw::redis::Redis>(connection_options);


    {
        std::unique_lock<std::mutex> lk(_workLock);
         _workRun = true;
        _workCond.notify_all();
    }

    do 
    {
        time(&t_now);


        if(localWorkMsg.size() <= 0){
            std::unique_lock<std::mutex> lk(_workLock);
            if(_lstWorkMsg.size() <= 0){
                _workCond.wait_until(lk, chrono::steady_clock::now() + std::chrono::seconds(10));
                continue;
            }
            localWorkMsg = std::move(_lstWorkMsg);
            continue;
        }
   
        if(ctx && ((t_now - t_last_get) > 30)){
            std::unordered_map<uint32_t,std::shared_ptr<ObjStatus>> obj_status;
            t_last_get = t_now;
            {
                std::unique_lock<std::mutex> vlck(_vlock);
                std::unique_lock<std::mutex> alck(_alock);
                for(auto iter = _proxyObjs.begin(); iter != _proxyObjs.end();iter++) 
                {
                    auto status = std::make_shared<ObjStatus>();
                    obj_status[iter->second->uid] = status;
                }     
            }

            for(auto iter = obj_status.begin(); iter != obj_status.end();iter++)
            {
                Json::Value root;
                Json::Reader reader;
                std::string config;
                std::stringstream sslock,sskey;

                sslock << "prop.persist."<< (uint32_t)iter->first << ".config.lock";
                sskey << "prop.persist."<< (uint32_t)iter->first << ".config";
                config = redisXXHashGetFieldLock(redis_client,sslock.str(),sskey.str(),"value");
                if(!reader.parse(config,root))
                    continue;
                //std::cout << "config_"<< iter->first<< "[ "  << config << " ]"<< std::endl;

                if(root.isMember("filerecord") && root["filerecord"].isInt())
                    iter->second->record_en = root["filerecord"].asInt() ;

                if(root.isMember("file_record") && root["file_record"].isInt())
                    iter->second->record_en = root["file_record"].asInt();
            }


            for(auto iter = obj_status.begin(); iter != obj_status.end();iter++)
            {
                Json::Value root;
                Json::Reader reader;
                std::string status;
                uint32_t enable = 0;
                std::stringstream sslock,sskey;
                sslock << "prop.status."<< (uint32_t)iter->first << ".value.lock";
                sskey << "prop.status."<< (uint32_t)iter->first << ".value";
                status = redisXXHashGetFieldLock(redis_client,sslock.str(),sskey.str(),"latest");
                if(!reader.parse(status,root))
                    continue;
                //std::cout << "status_"<< iter->first<< "[ "  << status << " ]"<< std::endl;

                if(root.isMember("signal") && root["signal"].isInt())
                     iter->second->rssi = root["signal"].asInt();

                if(root.isMember("battery") && root["battery"].isInt())
                     iter->second->battery = root["battery"].asInt();

                if(root.isMember("temperature") && root["temperature"].isInt())
                     iter->second->temperature = root["temperature"].asInt();     

                if(root.isMember("storage") && root["storage"].isArray()){
                    Json::Value array = root["storage"];
                    for (int i = 0;i<array.size();i++)
                    {
                        Json::Value item = array[i];
                        if(!item.isMember("free") || !item.isMember("total"))
                            continue;
                        if(!item["free"].isDouble() || !item["total"].isDouble())  
                            continue;
                        iter->second->disk_free = item["free"].asDouble() * 1024 * 1024;
                        iter->second->disk_total = item["total"].asDouble() * 1024 * 1024;
                        break;
                    }
                }
            }


            {
                std::unique_lock<std::mutex> vlck(_vlock);
                std::unique_lock<std::mutex> alck(_alock);
                for(auto iter = _proxyObjs.begin(); iter != _proxyObjs.end();iter ++)
                {
                    auto it = obj_status.find(iter->second->uid);
                    if(it != obj_status.end()){
                         //std::cout << "gb_status_"<< iter->first
                         //<< ",battery:"  << it->second->battery 
                         //<< ",rssi:"<< it->second->rssi 
                         //<< ",disk_free:"<< it->second->disk_free 
                         //<< ",disk_total:"<< it->second->disk_total 
                         //<< ",temperature:"<< it->second->temperature 
                         //<< ",record_en:"<< it->second->record_en 
                         //<< std::endl;
                        iter->second->status = *(it->second);
                    }
                }      
            }
        }
    

  
     
        for(auto msg : localWorkMsg)
        {
            Json::Value root;
            Json::Reader reader;
            std::string topic = msg.first;

            //std::cout << "topic:"<< topic
            //             << ",msg:"  << msg.second
            //             << std::endl;

            if(!reader.parse(msg.second,root))
                continue;            
           if (root.isMember("notify") && root["notify"].isString()) {
                std::string notify = root["notify"].asString();
                if (notify == "gb28181.export")  {
                    if(!root.isMember("params") || !root["params"].isObject())
                        continue;
                    
                    Json::Value& params = root["params"];
                    if (!params.isMember("export") || !params["export"].isArray() )
                        continue;
                    
                    Json::Value& exports = params["export"];                
                    if (exports.isArray()) {
                        uint32_t uid;
                        std::map<uint32_t,Json::Value> map_ids;
                        std::unique_lock<std::mutex> vlck(_vlock);
                        std::unique_lock<std::mutex> alck(_alock);
                        for (int32_t i  = 0 ;i < exports.size(); i++)
                        {
                
                            uid = exports[i]["gid"].asInt();
                            map_ids[uid] = exports[i];
                            auto iter = _proxyObjs.find(uid);
                            if (iter != _proxyObjs.end())
                                continue;
                            auto obj = std::make_shared<ProxyObj>();

                            obj->uid = exports[i]["gid"].asInt();
                            obj->on_pos = false;
                            obj->living = false;
                            obj->channel_online = false;
                            obj->index = _index++;
                            obj->name = exports[i]["dev_alias"].asString();
                            obj->gb28181_id = exports[i]["gb28181_id"].asUInt();

                            char deviceID[GBT_UNICODE_LENGTH + 1] = { 0 };
                            //memcpy(deviceID, m_server.username, 10);
                            //snprintf(deviceID + 10,sizeof(deviceID) - 10,"132%07d", obj->gb28181_id);   


                            memcpy(deviceID, m_server.username,10); 
                            snprintf(deviceID + 10,sizeof(deviceID) - 10,"13255%05d", obj->gb28181_id);

                            obj->channel = std::make_shared<GBGwChannel>(this,obj->name, m_server.username,std::string(deviceID), obj->index); 
                            std::cout << "new obj[" << _index << "] uid:"<< obj->uid << ",name:" << obj->name << ",gb28181_id:" << obj->gb28181_id << ",deviceID:"<< deviceID << ",channel:" << obj->channel<< std::endl;
                            _proxyObjs[obj->uid] = obj;
                        }

                        for(auto iter = _proxyObjs.begin(); iter != _proxyObjs.end();)
                        {
                            if(map_ids.end() == map_ids.find(iter->first)){
                                if(iter->second->on_pos){
                                    _terminal->offPosition(iter->second->uid);
                                    iter->second->on_pos = false;
                                }
                                if(iter->second->channel)
                                    iter->second->channel->offline();
                                std::cout << "free obj uid:"<<  iter->second->uid << ",name:" << iter->second->name  << ",gb28181_id:" << iter->second->gb28181_id << ",channel:" << iter->second->channel<< std::endl;                                                                     
                                iter = _proxyObjs.erase(iter);
                            } else {
                                iter++;
                            }
                        }
                        std::cout << "obj total:"<<  _proxyObjs.size()<< std::endl;                                                                                                          
                    } 
                } else if (notify == "mediaserver.pushup") {
                    
                    if(!root.isMember("params") || !root["params"].isObject())  
                        continue;

                    Json::Value& params = root["params"];
                    if ( !params.isMember("id") 
                      || !params["id"].isNumeric() 
                      || !params.isMember("room") 
                      || !params["room"].isNumeric() 
                      || !params.isMember("avidx") 
                      || !params["avidx"].isNumeric() )
                        continue;           
                                     
                    uint32_t uid = params["id"].asUInt();
                    uint32_t room = params["room"].asUInt();
                    uint32_t avidx = params["avidx"].asUInt();

                    if(uid != room || 0 != avidx)
                        continue;

                    {
                        std::unique_lock<std::mutex> vlck(_vlock);
                        std::unique_lock<std::mutex> alck(_alock);
                        auto it = _proxyObjs.find(uid);
                        if(it != _proxyObjs.end())
                            it->second->living = true;                       
                    }

                    {
                        std::unique_lock<std::mutex> lk(_connLock);
                        if(_connAbort || !_connRun)
                            continue;
                        _connCond.notify_all();
                    }
                
                } else if( ("login" == notify) && (SIGNAL_TOPIC_PUSH_SERVICE == topic)){
    
                    if(!root.isMember("params") || !root["params"].isObject())
                        continue;
                    
                    {
                        Json::Value identity,params;
                        params = root["params"];
                        if( !params.isMember("identity")  || !params["identity"].isObject())
                            continue;    
                        identity = params["identity"];
                        if(!identity.isMember("gid") || !identity["gid"].isInt())
                            continue;
                        {
                            std::unique_lock<std::mutex> vlck(_vlock);
                            std::unique_lock<std::mutex> alck(_alock);
                            uint64_t gid = identity["gid"].asInt();
                            auto iter = _proxyObjs.find(gid);
                            if(iter  == _proxyObjs.end())
                                continue;
                            if (!iter->second->channel_online) {
                                iter->second->channel_online = true;
                                if (iter->second->channel) 
                                    iter->second->channel->online();   
                                std::cout << "hannel->online() uid:"<<  iter->second->uid << ",name:" << iter->second->name  
                                        << ",gb28181_id:" << iter->second->gb28181_id << ",channel:" << iter->second->channel<< std::endl;       
                            }
                        }
                    }
                } else if(("logout" == notify) && (SIGNAL_TOPIC_PUSH_SERVICE == topic)){
     
                    if(!root.isMember("params") || !root["params"].isObject())
                        continue;
                    {
                        Json::Value identity,params;
                        params = root["params"];
                        if( !params.isMember("identity")  || !params["identity"].isObject())
                            continue;    
                        identity = params["identity"];
                        if(!identity.isMember("gid") || !identity["gid"].isInt())
                            continue;
            
                        {
                            std::unique_lock<std::mutex> vlck(_vlock);
                            std::unique_lock<std::mutex> alck(_alock);
                            uint64_t gid = identity["gid"].asInt();
                            auto iter = _proxyObjs.find(gid);
                            if(iter  == _proxyObjs.end())
                                continue;

                            if (iter->second->channel_online) {
                                iter->second->channel_online = false;
                                if (iter->second->channel) 
                                    iter->second->channel->offline();
                                std::cout << "hannel->offline() uid:"<<  iter->second->uid << ",name:" << iter->second->name  
                                        << ",gb28181_id:" << iter->second->gb28181_id << ",channel:" << iter->second->channel<< std::endl;
                            }
                        }
                    }
                }  else if( ("clients" == notify) && (SIGNAL_TOPIC_PUSH_SERVICE == topic)){

                    if(!root.isMember("params") || !root["params"].isObject())
                        continue;

                    {
                        std::vector<uint64_t> cli_ids;
                        Json::Value clients,params;
                        params = root["params"];
                        if( !params.isMember("clients")  || !params["clients"].isArray())
                            continue;    
                        clients = params["clients"];
                        for(size_t i = 0;i < clients.size(); i ++)
                        {
                            Json::Value cli = clients[i]; 
                            if(!cli.isMember("gid") || !cli["gid"].isInt())
                                continue;
                            cli_ids.push_back(cli["gid"].asInt());  
                        }

                        {
                            std::unique_lock<std::mutex> vlck(_vlock);
                            std::unique_lock<std::mutex> alck(_alock);
                            for( auto it : cli_ids)
                            {
                                auto iter = _proxyObjs.find(it);
                                if(iter  == _proxyObjs.end())
                                    continue;
                                if (!iter->second->channel_online) {
                                    iter->second->channel_online = true;
                                    if (iter->second->channel) 
                                        iter->second->channel->online();   
                                    std::cout << "hannel->online() uid:"<<  iter->second->uid << ",name:" << iter->second->name  
                                            << ",gb28181_id:" << iter->second->gb28181_id << ",channel:" << iter->second->channel<< std::endl;       
                                }
                            }
                        }
                    }
                }
            }
        }
        localWorkMsg.clear();
    }while(!_workAborth);

    if(ctx)
        redisFree(ctx);
    {
        std::unique_lock<std::mutex> lk(_workLock);
        _workCond.notify_all();
        _workRun = false;
    }
}

void GBGwServer::subThreadEntry()
{
    sw::redis::ConnectionOptions connection_options;
    connection_options.host = _redis_ip;  // Required.
    connection_options.port = _redis_port; // Optional. The default port is 6379.
    connection_options.password = MQSERVERPASS;
    connection_options.socket_timeout = std::chrono::milliseconds(REDIS_TIMEDOUT_MS);

    {
        std::unique_lock<std::mutex> lck0(_subLock);
        _subCond.notify_all();
        _subRun = true;
    }

    auto redis_subsciber = std::make_shared<sw::redis::Redis>(connection_options);
    auto work_subsriber = redis_subsciber->subscriber();
    work_subsriber.psubscribe({
        SIGNAL_TOPIC_PUBLIC_REQUEST".*",
        MEDIA_TOPIC_PUBLIC_NOTIFY,
        SIGNAL_TOPIC_PUSH_SERVICE,
        TOPCI_GB_DEFAULT
    });

    auto redis_message_handler = [&] (std::string pattern, std::string channel, std::string msg){
        std::unique_lock<std::mutex> lk(_workLock);
        _lstWorkMsg.push_back(std::make_pair(channel,msg));
        _workCond.notify_all();
    };
      
    work_subsriber.on_message([redis_message_handler](std::string channel, std::string msg) {
        redis_message_handler("", channel, msg);
    });

    work_subsriber.on_pmessage([redis_message_handler](std::string pattern, std::string channel, std::string msg) {
        redis_message_handler(pattern, channel, msg);
    });

    work_subsriber.on_meta([](sw::redis::Subscriber::MsgType type, sw::redis::OptionalString channel, long long num) {
        // Process message of META type.
    });


    do{   
        try {
            work_subsriber.consume();
        } catch (const sw::redis::Error &err) {
            // Handle exceptions.
        }
    }while(!_subAborth);

    {
        std::unique_lock<std::mutex> lck0(_subLock);
        _subCond.notify_all();
        _subRun = false;
    }
}

void GBGwServer::connThreadEntry()
{
    bool termConn = false;
    bool termAbort = false;
    {
        std::unique_lock<std::mutex> lck0(_connLock);
        _connCond.notify_all();
        _connRun = true;
        termConn = false;
    }

    do
    {
        if(!termConn){
            termConn = _terminal->connect(100,_media_ip,_media_port);
            std::cout << "terminal connect:"<< termConn << std::endl;
        } 

        if(termConn){
          
            {
                std::unique_lock<std::mutex> vlck(_vlock);
                std::unique_lock<std::mutex> alck(_alock);
                for(auto iter = _proxyObjs.begin(); iter != _proxyObjs.end();iter++)
                {
                    auto second = iter->second;
    
                    if(second->on_pos || !second->living)
                        continue;
                    if(!_terminal->onPosition(second->uid))
                        continue; 
                    second->on_pos = true;
                }
            }
            
          
            {
                std::unique_lock<std::mutex> lk(_connLock);
                _termAbort = false;
                //_connCond.wait_until(lk,std::chrono::steady_clock::now() + std::chrono::seconds(5));
                _connCond.wait(lk);
                termAbort = _termAbort;
            }

            if(termAbort){
                termConn = false;
                std::unique_lock<std::mutex> vlck(_vlock);
                std::unique_lock<std::mutex> alck(_alock);
                _terminal->disconnect();
                for(auto iter:_proxyObjs){
                    iter.second->on_pos = false;
                    iter.second->living = false;
                }   
            }
        } else {
            //连接失败 超时重连
            std::cout << "terminal connect failed try again 5s later" << std::endl;
            std::unique_lock<std::mutex> lk(_connLock);
            _connCond.wait_until(lk,std::chrono::steady_clock::now() + std::chrono::seconds(5));
        }
    }while(!_connAbort);

    termConn = false;
    _terminal->disconnect();


    {
        std::unique_lock<std::mutex> lck0(_connLock);
        _connCond.notify_all();
        _connRun = false;
    }
}


void GBGwServer::onRelayConnectionLost()
{
    {
        std::unique_lock<std::mutex> lck0(_connLock);
        if(_connAbort || !_connRun)
            return;
         _termAbort = true;
        _connCond.notify_all();
    }
}

void GBGwServer::onRelayAudioData(uint64_t id,char *data,uint32_t length,uint64_t  pts)
{
    std::unique_lock<std::mutex> alck(_alock);
    auto iter = _proxyObjs.find(id);
    if(iter  == _proxyObjs.end())
        return;
    if (!iter->second->channel_online)
    {
        if (iter->second->channel)
        {
            iter->second->channel->online();
            iter->second->channel_online = true;
            std::cout << "channel->online(): " << id << std::endl;
        }
    }
    //std::cout << "onRelayAudioData >>>> uid:" << id <<",length:"<< length << ",pts" << pts << std::endl;

    if(iter->second->channel)
        iter->second->channel->writeAudio(data,length,pts * 1000);
        
}

void GBGwServer::onRelayVideoData(uint64_t id, char *data, uint32_t length,uint64_t  pts, bool key, bool hevc)
{
    std::unique_lock<std::mutex> vlck(_vlock);
    auto iter = _proxyObjs.find(id);
    if(iter  == _proxyObjs.end())
        return;
    if (!iter->second->channel_online)
    {
        if (iter->second->channel)
        {
            iter->second->channel->online();
            iter->second->channel_online = true;
            std::cout << "video channel->online(): " << id << std::endl;
        }

    }

    //std::cout << "onRelayVideoData >>>> uid:" << id <<",length:"<< length << ",pts" << pts << std::endl;

    if(iter->second->channel)
        iter->second->channel->writeVideo(data,length,pts * 1000,key,hevc);
}


void GBGwServer::onRelayProviderOnPosition(uint64_t uid, uint64_t room, uint64_t index)
{
    //std::cout << "onRelayProviderOnPosition,uid:"<< uid <<",room:" << room << ",index"<< (uint32_t) index<< std::endl;
    if((uid != room) || (0 != index)){
        return;
    }

    {
        std::unique_lock<std::mutex> vlck(_vlock);
        std::unique_lock<std::mutex> alck(_alock);
        auto iter = _proxyObjs.find(uid);
        if(iter  == _proxyObjs.end())
            return;
        if (iter->second->channel && !iter->second->channel_online){
            iter->second->channel->online();
            iter->second->channel_online = true;
        }

    }
    std::cout << "onRelayProviderOnPosition uid:" << uid << ",room:" << room << ",index:" << (uint32_t)index << std::endl;
}

void GBGwServer::onRelayProviderOffPosition(uint64_t uid, uint64_t room, uint64_t index)
{
    if((uid != room) || (0 != index))
        return;
    #if 0
    //std::cout << "onRelayProviderOffPosition,uid:"<< uid <<",room:" << room << ",index"<< (uint32_t) index<< std::endl;
    {
        std::unique_lock<std::mutex> vlck(_vlock);
        std::unique_lock<std::mutex> alck(_alock);
        auto iter = _proxyObjs.find(uid);
        if(iter  == _proxyObjs.end())
            return;
        if (iter->second->channel && iter->second->channel_online)
        {
            iter->second->channel->offline();
            iter->second->channel_online = false;
        }
    }
    #endif
    std::cout << "onRelayProviderOffPosition,uid:"<< uid <<",room:" << room << ",index:"<< (uint32_t) index<< std::endl;
}



//
void GBGwServer::onConnectStatus(int status, int64_t clock)
{

}

void GBGwServer::handlePtzCommand(int id, const char* deviceID, const GBT_PtzCommand& cmd)
{

}

void GBGwServer::handleTeleBoot(int id, const char* deviceID, GBT_TeleBoot& boot)
{
    uint64_t uid = 0;
    Json::RawWriter writer;
    Json::Value root,params,config;
    std::map<uint64_t,std::shared_ptr<ProxyObj>> proxyObjs;

    {
        std::unique_lock<std::mutex> vlck(_vlock);
        std::unique_lock<std::mutex> alck(_alock);
        proxyObjs = _proxyObjs;
    }

    for(auto iter = proxyObjs.begin(); iter != proxyObjs.end();iter++)
    {
        auto second = iter->second;
        if(id == second->index){
            uid = second->uid;
            break;
        }
    }
    
    std::cout << __FUNCTION__ << ",at:"<< __LINE__ << ",uid:"<< uid << ",id:"<< id << ",deviceid:" <<deviceID <<std::endl;
    if(0 == uid)
        return;

    root["cseq"] = Json::Value(0);
    root["to_gid"] = Json::Value((uint32_t)uid);
    root["method"] = Json::Value("remote.restart");

    root["auth"]["gid"] = Json::Value((int)uid);
    root["auth"]["term"] = Json::Value((int)TERM_TYPE_GB28181_SVR);
    root["auth"]["ip"] = Json::Value("127.0.0.1");
    root["auth"]["key"] = Json::Value("");
    root["auth"]["login_code"] = Json::Value(0);

    postPubMessage("signal.request.0",writer.write(root));
}

void GBGwServer::handleForceKeyFrame(int id, const char* deviceID)
{
 
}

void GBGwServer::handleZoom(int id, const char* deviceID, GBT_DragZoomMode zoomMode, const GBT_DragZoom& dragZoom)
{

}

/// 有应答命令.
bool GBGwServer::handleRecordCmd(int id, const GBT_Command& req, const GBT_RecordCmd& cmd)
{

    uint64_t uid = 0;
    Json::RawWriter writer;
    Json::Value root,params,config;
    std::map<uint64_t,std::shared_ptr<ProxyObj>> proxyObjs;

    {
        std::unique_lock<std::mutex> vlck(_vlock);
        std::unique_lock<std::mutex> alck(_alock);
        proxyObjs = _proxyObjs;
    }

    for(auto iter = proxyObjs.begin(); iter != proxyObjs.end();iter++)
    {
        auto second = iter->second;
        if(id == second->index){
            uid = second->uid;
            break;
        }
    }

    if(0 == uid)
        return false;

    root["cseq"] = Json::Value(0);
    root["to_gid"] = Json::Value((uint32_t)uid);
    root["method"] = Json::Value("config.set");

    if(cmd.cmd == GBT_START_RECORD)
    {
        config["filerecord"] =  Json::Value(1);
    }
    else if(cmd.cmd == GBT_STOP_RECORD)
    {
        config["filerecord"] =  Json::Value(0);
    }

    params["config"] = config;
    root["params"] = params;

    root["auth"]["gid"] = Json::Value((int)uid);
    root["auth"]["term"] = Json::Value((int)TERM_TYPE_GB28181_SVR);
    root["auth"]["ip"] = Json::Value("127.0.0.1");
    root["auth"]["key"] = Json::Value("");
    root["auth"]["login_code"] = Json::Value(0);

    postPubMessage("signal.request.0",writer.write(root));
    return  true;
}

bool  GBGwServer::handleGuardCmd(int id, const GBT_Command& req, const GBT_GuardCmd& cmd)
{
    return  false;
}

bool  GBGwServer::handleAlarmCmd(int id, const GBT_Command& req, const GBT_AlarmCmd& cmd)
{
    return  false;
}

bool  GBGwServer::handleSetHomePosition(int id, const GBT_Command& req, const GBT_HomePosition& cmd)
{
    return  false;
}

bool  GBGwServer::handleDeviceConfig(int id, const GBT_Command& req, const GBT_DeviceConfig& config)
{
    uint64_t uid = 0;
    Json::RawWriter writer;
    Json::Value root,params,jconfig;
    std::map<uint64_t,std::shared_ptr<ProxyObj>> proxyObjs;

    int hb_count = config.basicParam.heartBeatCount;
    int hb_interval = config.basicParam.heartBeatInterval;
    getAgent().setHeartBeat(hb_count,hb_interval);


    {
        std::unique_lock<std::mutex> vlck(_vlock);
        std::unique_lock<std::mutex> alck(_alock);
        proxyObjs = _proxyObjs;
    }

    for(auto iter = proxyObjs.begin(); iter != proxyObjs.end();iter++)
    {
        auto second = iter->second;
        if(id == second->index){
            uid = second->uid;
            break;
        }
    }
    if(0 == uid)
        return false;

    root["cseq"] = Json::Value(0);
    root["to_gid"] = Json::Value((uint32_t)uid);
    root["method"] = Json::Value("config.set");

    switch(config.videoParam.resolution)
    {
        case 5: //720P
            jconfig["videoresolution"] =  Json::Value("1280x720");
            break;
        case 6: //1080P
            jconfig["videoresolution"] =  Json::Value("1920x1080");
            break;
        case 1: //QCIF
            jconfig["videoresolution"] =  Json::Value("176x114");
            break;
        case 2: //CIF
             jconfig["videoresolution"] =  Json::Value("352x288");
            break;
        case 4: //D1
             jconfig["videoresolution"] =  Json::Value("720x576");
            break;
        case 7: //4K
            break;
    }

    params["config"] = jconfig;
    root["params"] = params;

    root["auth"]["gid"] = Json::Value((int)uid);
    root["auth"]["term"] = Json::Value((int)TERM_TYPE_GB28181_SVR);
    root["auth"]["ip"] = Json::Value("127.0.0.1");
    root["auth"]["key"] = Json::Value("");
    root["auth"]["login_code"] = Json::Value(0);

    postPubMessage("signal.request.0",writer.write(root));
    return  true;
}


/// 查询.
bool  GBGwServer::handleQueryDeviceInfo(int id, const GBT_Command& req, GBT_DeviceInfo& info)
{
    uint64_t uid = 0;
    bool fined = false;
    std::map<uint64_t,std::shared_ptr<ProxyObj>> proxyObjs;

    {
        std::unique_lock<std::mutex> vlck(_vlock);
        std::unique_lock<std::mutex> alck(_alock);
        proxyObjs = _proxyObjs;
    }

  
    for(auto iter = proxyObjs.begin(); iter != proxyObjs.end();iter++)
    {
        auto second = iter->second;
        if(!second->channel || second->index != id || !second->channel_online)
            continue;
        fined = true;
        auto catalog = second->channel->getCatalog();
        strcpy(info.name,catalog.name);
        strcpy(info.manufacturer, catalog.manufacturer);
        strcpy(info.model, catalog.model);
        strcpy(info.firmware, "EasyGbSvr");
        info.channel = _proxyObjs.size();
        std::cout << "device info: " << catalog.name  <<std::endl;
    }
    return  fined;
}

bool  GBGwServer::handleQueryDeviceStatus(int id, const GBT_Command& req, GBT_DeviceStatus& status)
{
    bool fined = false;
    std::map<uint64_t,std::shared_ptr<ProxyObj>> proxyObjs;

    {
        std::unique_lock<std::mutex> vlck(_vlock);
        std::unique_lock<std::mutex> alck(_alock);
        proxyObjs = _proxyObjs;
    }

    for(auto iter = proxyObjs.begin(); iter != proxyObjs.end();iter++)
    {
        auto second = iter->second;
        if(!second->channel || second->index != id || !second->channel_online)
            continue;
        fined = true;
        auto catalog = second->channel->getCatalog();
        fined = true;
        status.status = GBT_STATUS_ON;
        status.encode = GBT_ON;
        status.record = second->status.record_en ? GBT_ON: GBT_OFF;

        status.rssi = second->status.rssi;
        status.diskTotal = second->status.disk_total;
        status.disFree = second->status.disk_free ;
        status.battery = second->status.battery ;
        status.temperature = second->status.temperature ;
        
        status.deviceTime = gbtTimeNow();
    }
    std::cout << __FUNCTION__ << ",at:"<< __LINE__ << ",fined:"<< fined << ",id:"<< id << ",deviceid:" << req.deviceID  <<std::endl;
    if(!fined)
       std::cout << "outonline: " << req.deviceID <<std::endl;
    status.online = fined ? GBT_ONLINE:GBT_OFFLINE;
    return  true;
}


bool  GBGwServer::handleQueryBasicParam(int id, const GBT_Command& req, GBT_BasicParam& config)
{

    //char name[GBT_NAME_LENGTH];         /// 名称.

    //char deviceID[GBT_UNICODE_LENGTH + 1];
    //char serverID[GBT_UNICODE_LENGTH + 1];
    //char serverIP[GBT_IP_LENGTH];
    //int  serverPort;
    //char domainName[GBT_NAME_LENGTH];
    //int expiration;   ///
    //int heartBeatInterval;
    //int heartBeatCount;
    //int positionCapability;
    //double longitude;
    //double latitude;

    return  false;
}

bool  GBGwServer::handleQueryVideoParamOpt(int id, const GBT_Command& req, GBT_VideoParamOpt& config)
{
    return false;
}
bool  GBGwServer::handleQueryVideoParamConfig(int id, const GBT_Command& req, GBT_VideoParamConfig& config)
{
    return false;
}
bool  GBGwServer::handleQueryAudioParamOpt(int id, const GBT_Command& req, GBT_AudioParamOpt& config)
{
    return false;
}
bool  GBGwServer::handleQueryAudioParamConfig(int id, const GBT_Command& req, GBT_AudioParamConfig& config)
{
    return false;
}

int  GBGwServer::handleQueryRecordInfo(int id, const GBT_Command& req, const GBT_RecordInfoQuery& query, GBT_RecordInfo* records, int size)
{
    //query.deviceID
    //query.startTime
    //query.endTime;
    return  0;
}

int  GBGwServer:: handleQueryAlarm(int id, const GBT_Command& req, const GBT_AlarmQuery& query, GBT_Alarm* alarms, int size)
{
    return  0;
}

int  GBGwServer:: handleQueryPreset(int id, const GBT_Command& req, const GBT_PresetQuery& query, GBT_Preset* presets, int size)
{
    return  0;
}

bool GBGwServer::handleSW(int id, const GBT_Command& req, const GBT_DeviceSwitch& sw)
{

    uint64_t uid = 0;
    Json::RawWriter writer;
    Json::Value root,params,config;
    std::map<uint64_t,std::shared_ptr<ProxyObj>> proxyObjs;

    {
        std::unique_lock<std::mutex> vlck(_vlock);
        std::unique_lock<std::mutex> alck(_alock);
        proxyObjs = _proxyObjs;
    }

    for(auto iter = proxyObjs.begin(); iter != proxyObjs.end();iter++)
    {
        auto second = iter->second;
        if(id == second->index){
            uid = second->uid;
            break;
        }
    }
    if(0 == uid)
        return false;


    root["cseq"] = Json::Value(0);

    root["to_gid"] = Json::Value((uint32_t)uid);
    if(0 == sw.index){
        root["method"] = Json::Value(sw.enable ? "ptz.up" : "ptz.down");
    } else  if(1 == sw.index){
        root["method"] = Json::Value(sw.enable ? "ptz.left" : "ptz.right");
    } else  if(2 == sw.index){
        root["method"] = Json::Value(sw.enable ? "ptz.zin" : "ptz.zout");
    } else  if(3 == sw.index){
        //开 关
        root["method"] = Json::Value(sw.enable ? "ptz.fin" : "ptz.fout");
    } else  if(4 == sw.index){
        //manual  //auto
        root["method"] = Json::Value(sw.enable ? "ptz.winwiper.on" : "ptz.winwiper.off");
    } else  if(10 == sw.index){
        root["method"] = Json::Value("ptz.focus");    
    }
   

    root["auth"]["gid"] = Json::Value((int)uid);
    root["auth"]["term"] = Json::Value((int)TERM_TYPE_GB28181_SVR);
    root["auth"]["ip"] = Json::Value("127.0.0.1");
    root["auth"]["key"] = Json::Value("");
    root["auth"]["login_code"] = Json::Value(0);

    postPubMessage("signal.request.0",writer.write(root));

    return  true;
}
bool  GBGwServer::handleQueryMobilePosition(int id, const GBT_Command& req, const GBT_MobilePositionQuery& query)
{
    return  false;
}

//custom defined message
bool GBGwServer::handleMessage(int id, const GBT_Command& req, GBT_Command& resp)
{
    return  false;
}

int GBGwServer::openPlayback(const GBT_RecordUrl &record, double speed, MStreamSource **ppStream)
{
    return  -1;
}

void GBGwServer::closePlayback(MStreamSource *stream)
{

}


void GBGwServer::onBeginPlay(uint32_t index)
{
    uint64_t uid = 0;
    Json::RawWriter writer;
    Json::Value root,params,config;
    std::map<uint64_t,std::shared_ptr<ProxyObj>> proxyObjs;
    //std::cout << __FUNCTION__ << ",at:"<< __LINE__ << ",index:"<< index <<std::endl;
    {
        std::unique_lock<std::mutex> vlck(_vlock);
        std::unique_lock<std::mutex> alck(_alock);
        proxyObjs = _proxyObjs;
    }

    for(auto iter = proxyObjs.begin(); iter != proxyObjs.end();iter++)
    {
        auto second = iter->second;
        if(index == second->index){
            uid = second->uid;
            break;
        }
    }

    if(0 == uid)
        return ;

    root["cseq"] = Json::Value(0);
    root["to_gid"] = Json::Value((uint32_t)uid);
    root["from_gid"] = Json::Value((uint32_t)1000000000);
    root["method"] = Json::Value("live.invite");

    params["config"] = config;
    root["params"] = params;

    root["auth"]["gid"] = Json::Value((int)uid);
    root["auth"]["term"] = Json::Value((int)TERM_TYPE_GB28181_SVR);
    root["auth"]["ip"] = Json::Value("127.0.0.1");
    root["auth"]["key"] = Json::Value("");
    root["auth"]["login_code"] = Json::Value(0);

    std::cout << __FUNCTION__ << ",at:"<< __LINE__ << ",index:"<< index <<std::endl;
    postPubMessage("signal.request.0",writer.write(root));

}


void GBGwServer::onEndPlay(uint32_t index)
{
    uint64_t uid = 0;
    Json::RawWriter writer;
    Json::Value root,params,config;
    std::map<uint64_t,std::shared_ptr<ProxyObj>> proxyObjs;

    {
        std::unique_lock<std::mutex> vlck(_vlock);
        std::unique_lock<std::mutex> alck(_alock);
        proxyObjs = _proxyObjs;
    }

    for(auto iter = proxyObjs.begin(); iter != proxyObjs.end();iter++)
    {
        auto second = iter->second;
        if(index == second->index){
            uid = second->uid;
            break;
        }
    }

    if(0 == uid)
        return;

    root["cseq"] = Json::Value(0);
    root["to_gid"] = Json::Value((uint32_t)uid);
    root["from_gid"] = Json::Value((uint32_t)1000000000);
    root["method"] = Json::Value("live.invite.cancel");
    root["params"] = params;

    root["auth"]["gid"] = Json::Value((int)uid);
    root["auth"]["term"] = Json::Value((int)TERM_TYPE_GB28181_SVR);
    root["auth"]["ip"] = Json::Value("127.0.0.1");
    root["auth"]["key"] = Json::Value("");
    root["auth"]["login_code"] = Json::Value(0);
    
    std::cout << __FUNCTION__ << ",at:"<< __LINE__ << ",index:"<< index <<std::endl;
    postPubMessage("signal.request.0",writer.write(root));
}

